unit uCxtNet;

interface

uses
  SysUtils, Classes, Windows, uDZSysUtils, uDZStream, uDZTrace, uDZPersistent,
  uDZThread, uDZSocket, uDZWinsock2, uDZIOCP, uDZCRC,
  uCxtTypes, uCxtPackage, uAppConfig, uAppLog, uSocket, uCxtSession;

{.$define output_request_xml}

type
  TRequestState = (rsFixedHeader, rsVariableHeader, rsContent);
  TCxtConnection = class(TServerConnection)
  private
    fConnectionID: AnsiString;
    ReqHeader: TPackageHeader;
    Content: TMemoryStream; 
    State: TRequestState;
    fSession: TCxtSession;
  private
    function GetUserName: string;
    procedure ReadFixedHeader;
    procedure ReadVariableHeader;
    procedure ReadContent;
  public
    constructor Create; override;
    destructor Destroy; override;
    procedure OnConnect;
    procedure OnFixedHeader;
    procedure OnVariableHeader;
    procedure OnContent;
    property ConnectionID: AnsiString read fConnectionID;
    property Session: TCxtSession read fSession write fSession;
    property UserName: string read GetUserName;
  end;

  TCxtAcceptor = class(TDzCustomServerSocket)
  public
    function Close: Boolean; override;
  end;
  
  TCxtPkgProcesser =  procedure(Conn: TCxtConnection; ReqType: TRequestType; 
    Req: TRequestPackage);

function PostAsyncAccept(Acceptor: TCxtAcceptor; Number: Cardinal): Cardinal;

var
  PkgProcessor: TCxtPkgProcesser;
  
implementation

procedure AcceptCallback(Sender: PDzIocpIOService; Socket: TDzCustomServerSocket;
  Operation: PDzAsyncAccept); forward;

function AsyncAccept(Acceptor: TDzCustomServerSocket;
  Connection: TDzServerConnection): Boolean;
var
  LastError: Integer;
begin
  Result := Acceptor.AsyncAccept(Connection, @AcceptCallback, @LastError);
  if not Result then
  begin
    FmtWriteLog(tsError, 'accept error: %d'#10, [LastError]);
  end;
end;

function PostAsyncAccept(Acceptor: TCxtAcceptor; Number: Cardinal): Cardinal;
var
  i: LongWord;
  Connection: TCxtConnection;
begin
  Result := 0;
  for i := 1 to Number do
  begin
    Connection := TCxtConnection.Create;
    if AsyncAccept(Acceptor, Connection) then Inc(Result)
    else begin
      Connection.Free;
      Break;
    end;
  end;
end;

procedure AcceptCallback(Sender: PDzIocpIOService; Socket: TDzCustomServerSocket;
  Operation: PDzAsyncAccept);
var
  BytesTrans, Flags: DWORD;
  LastError: Integer;
begin
  try
    if WSAGetOverlappedResult(Operation^.Connection.Handle,
      POverlapped(Operation), BytesTrans, False, Flags) then
    with TCxtConnection(Operation.Connection) do
    begin
      UpdateAcceptContext(Socket);
      PostAsyncAccept(TCxtAcceptor(Socket), 1);
      Sender.Bind(Handle, Operation.Connection);
      OnConnect;
    end
    else begin
      LastError := WSAGetLastError;
      FmtWriteLog(tsError, 'accept error: %d'#10, [LastError]);
      if LastError = ERROR_OPERATION_ABORTED then
      begin
        Operation.Connection.Close;
        Socket.Close;
      end
      else begin
        if AsyncAccept(Socket, Operation.Connection) then
        begin
          Operation.Connection.Close;
          Socket.Close;
        end;
      end;
    end;
  finally
    Socket.DeallocAcceptOperation(Operation);
    Socket.Release;
  end;
end;

procedure RecvCallback(Sender: TDzIocpIOService;
  Socket: TDzCustomConnection;
  Operation: PDzAsyncRecv);
var
  BytesTrans, Flags: DWORD;
  Connection: TCxtConnection absolute Socket;
begin
  try
    if not WSAGetOverlappedResult(Connection.Handle,
      POverlapped(Operation), BytesTrans, False, Flags) then
    begin
      FmtWriteLog(tsError, 'recv error: %d'#10, [WSAGetLastError]);
      Socket.Close;
      Exit;
    end;
    if BytesTrans = 0 then
    begin
      Socket.Close;
      WriteLog(tsError, 'connection closed gracefully'#10);
      Exit;
    end;
    case Connection.State of
      rsFixedHeader: Connection.OnFixedHeader;
      rsVariableHeader: Connection.OnVariableHeader;
      rsContent: Connection.OnContent;
    end;
  finally
    Connection.DeallocRecvOperation(Operation);
    Socket.Release;
  end;
end;

{ TCxtConnection }

constructor TCxtConnection.Create;
begin
  inherited;
  Content := TMemoryStream.Create;
end;

destructor TCxtConnection.Destroy;
begin
  Content.Free;
  inherited;
end;

function TCxtConnection.GetUserName: string;
begin 
  if Assigned(fSession) then Result := fSession.UserName
  else Result := '';
end;

procedure TCxtConnection.OnFixedHeader;
begin
  if not ReqHeader.Fixed.IsValidRequest then 
  begin
    WriteLog(tsDebug, 'invalid request header, close connection'#10);
    Close;
    Exit;
  end;
  if ReqHeader.Fixed.HdrLen > SizeOf(TFixedPackageHeader) then
  begin
    State := rsVariableHeader;
    ReadVariableHeader;
  end
  else begin
    State := rsContent;
    ReadContent;
  end;
end;

procedure TCxtConnection.OnVariableHeader;
begin
  if CRC32(@ReqHeader, ReqHeader.Fixed.HdrLen - 4) <> 
    PLongWord(LongInt(@ReqHeader) + ReqHeader.Fixed.HdrLen - 4)^ then
  begin
    WriteLog(tsError, 'crc check error, close connection'#10);
    Close;
    Exit;  
  end;
  State := rsContent;
  ReadContent;
end;

procedure TCxtConnection.OnConnect;
begin
  fConnectionID := Format('%s,%d', [RemoteAddr, RemotePort]);
  FmtWriteLog(tsDebug, 'accept: (%s, %d)'#10, [RemoteAddr, RemotePort]);
  Link;
  ReadFixedHeader;
end;

procedure TCxtConnection.OnContent;
var
  ReqType: TRequestType;
  Request: TRequestPackage;
begin
  ReqType := TRequestType(ReqHeader.Fixed.ReqType);

  FmtWriteLog(tsDebug, '%s request(%d bytes)'#10, [ReqTypeNames[ReqType],
    ReqHeader.Fixed.DataLen]);

  Request := RequestClasses[ReqType].Create;
  Content.Seek(0, soFromBeginning);
  Request.LoadFromStream(Content);
  {$ifdef output_request_xml}
  Writeln(Request.AsXML);
  {$endif}
  State := rsFixedHeader;
  Self.AddRef;
  PkgProcessor(Self, ReqType, Request);
  ReadFixedHeader;
end;

procedure TCxtConnection.ReadContent;
var
  LastError: Integer;
begin
  if Content.Size < ReqHeader.Fixed.DataLen then
    Content.Size := ReqHeader.Fixed.DataLen;

  if not AsyncRecv(Content.Memory^, ReqHeader.Fixed.DataLen, @RecvCallback,
    @LastError, MSG_WAITALL) then
  begin
    FmtWriteLog(tsError, 'recv error: %d'#10, [LastError]);
    Close;
  end;
end;

procedure TCxtConnection.ReadFixedHeader;
var
  LastError: Integer;
begin
  if not AsyncRecv(ReqHeader, SizeOf(TFixedPackageHeader), @RecvCallback,
    @LastError, MSG_WAITALL) then
  begin
    FmtWriteLog(tsError, 'recv error: %d'#10, [LastError]);
    Close;
  end;
end;

procedure TCxtConnection.ReadVariableHeader;
var
  LastError: Integer;
begin
  if not AsyncRecv(Pointer(Cardinal(@ReqHeader) + SizeOf(TFixedPackageHeader))^,
    ReqHeader.Fixed.HdrLen - SizeOf(TFixedPackageHeader),
    @RecvCallback, @LastError, MSG_WAITALL) then
  begin
    FmtWriteLog(tsError, 'recv error: %d'#10, [LastError]);
    Close;
  end;
end;

{ TCxtAcceptor }

function TCxtAcceptor.Close: Boolean;
begin
  Result := inherited Close;
  if Result then Release;
end;

end.
